package org.raymond.iworks.study.basic.thread.locksupport;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class LockSupportDemo {
    public static void main(String[] args) throws Exception {
        test3();
    }

    // pass
    public static void test1() throws InterruptedException {
        List<Thread> tList = new ArrayList<>();
        Runnable r1 = ()->{
            log.info("park...");
            tList.add(Thread.currentThread());
            LockSupport.park();
            log.info("unpark...");
        };
        Thread t1 = new Thread(r1);
        Thread t2 = new Thread(r1);
        log.info("new {}, {}", t1.getState().name(), t2.getState().name());
        t1.start();
        t2.start();
        log.info("run {}, {}", t1.getState().name(), t2.getState().name());

        Thread.sleep(1000);
        log.info("wait {}, {}", t1.getState().name(), t2.getState().name());

        Thread.sleep(3000);
        LockSupport.unpark(t1);
        log.info("unpark_t1 {}, {}", t1.getState().name(), t2.getState().name());

        Thread.sleep(1000);
        log.info("terminated_t1 {}, {}", t1.getState().name(), t2.getState().name());

        Thread.sleep(3000);
        LockSupport.unpark(t2);
        log.info("unpark_t2 {}, {}", t1.getState().name(), t2.getState().name());

        Thread.sleep(1000);
        log.info("terminated_t2 {}, {}", t1.getState().name(), t2.getState().name());
    }

    // 问题,如何判断线程已经执行到了park点?
    // 解决: 使用countdownlatch
    // pass
    public static void test2() throws InterruptedException {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2,
                0, TimeUnit.SECONDS, new LinkedBlockingQueue(), (r)->{
            Thread t = new Thread(r);
            return t;
        });
        CountDownLatch cdl = new CountDownLatch(2);
        List<Thread> tList = new ArrayList<>();
        Runnable r1 = ()->{
            log.info("park...");
            tList.add(Thread.currentThread());
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
            }
            // 模拟业务已完成操作,计数器-1
            cdl.countDown();
            // 进入休眠
            LockSupport.park();
            log.info("unpark...");
        };
        pool.execute(r1);
        pool.execute(r1);
        log.info("size:{}", tList.size());
        cdl.await();
        // Thread.sleep(5000);
        log.info("size:{}", tList.size());
        tList.forEach(LockSupport::unpark);
        pool.shutdown();
    }

    // CompletableFuture 无法判断何时执行到了park, 因为触发get后就阻塞在了park上,主线程无法执行await
    public static void test3() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2,
                0, TimeUnit.SECONDS, new LinkedBlockingQueue(), (r)->{
            Thread t = new Thread(r);
            return t;
        });
        AtomicBoolean rollback = new AtomicBoolean(false);
        List<Thread> tList = new ArrayList<>();
        CyclicBarrier cdl = new CyclicBarrier(3);
        Consumer<Boolean> consumer = (ex)-> {
            try {
                Thread.sleep(1000);
                if(ex){
                    throw new RuntimeException("throw ex");
                }
            } catch (InterruptedException | RuntimeException e) {
                log.error("{}", e.getMessage());
                rollback.set(true);
            }
        };
        Runnable r1 = newTxRunnable(consumer, false, rollback, cdl, tList);
        Runnable r2 = newTxRunnable(consumer, true, rollback, cdl, tList);
        pool.execute(r1);
        pool.execute(r2);

        try {
            log.info("await1 start {}", Thread.currentThread().getName());
            cdl.await();
            log.info("await1 end {}", Thread.currentThread().getName());
            log.info("await2 start {}", Thread.currentThread().getName());
            cdl.await();
            log.info("await2 end {}", Thread.currentThread().getName());
        } catch (InterruptedException | BrokenBarrierException e) {
        }

        log.info("shutdown");
        pool.shutdown();
    }

    public static Runnable newTxRunnable(Consumer<Boolean> apply, Boolean b,
                                  AtomicBoolean rollback, CyclicBarrier cyclicBarrier, List<Thread> threads){
        threads.add(Thread.currentThread());
        return ()->{
            apply.accept(b);
            try {
                log.info("await1 start {}", Thread.currentThread().getName());
                cyclicBarrier.await();
                log.info("await1 end {}", Thread.currentThread().getName());
            } catch (InterruptedException | BrokenBarrierException e) {
            }
            log.info("{} {}", Thread.currentThread().getName(), rollback.get());
            if(rollback.get()){
                log.info("rollback");
            }else{
                log.info("execute");
            }
            try {
                log.info("await2 start {}", Thread.currentThread().getName());
                cyclicBarrier.await();
                log.info("await2 end {}", Thread.currentThread().getName());
            } catch (InterruptedException | BrokenBarrierException e) {
            }
        };
    }
}
